package com.ice.remoting;

import java.net.InetSocketAddress;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;

import cn.hutool.core.lang.Pair;
import com.ice.framework.common.exception.BusinessException;
import com.ice.remoting.common.ThreadUtils;
import com.ice.remoting.data.DeviceLiveInfo;
import com.ice.remoting.exception.RemotingConnectException;
import com.ice.remoting.exception.RemotingSendRequestException;
import com.ice.remoting.exception.RemotingTimeoutException;
import com.ice.remoting.netty.*;
import lombok.SneakyThrows;
import org.apache.commons.lang3.exception.ExceptionUtils;

import com.ice.remoting.code.DecoderFactory;
import com.ice.remoting.code.EncodeFactory;
import com.ice.remoting.common.NetworkUtil;
import com.ice.remoting.common.RemotingHelper;
import com.ice.remoting.common.ThreadFactoryImpl;
import com.ice.remoting.data.RemotingCommand;
import com.ice.remoting.listener.ChannelEventListener;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.epoll.Epoll;
import io.netty.channel.epoll.EpollSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.DefaultEventExecutorGroup;
import io.netty.util.concurrent.EventExecutorGroup;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public abstract class NettyRemotingClient extends NettyRemotingAbstract implements DecoderFactory, EncodeFactory {

    private final Bootstrap bootstrap;
    private final EventLoopGroup eventLoopGroup;

    private NettyClientConfig nettyClientConfig;
    private final ExecutorService publicExecutor;
    private EventExecutorGroup defaultEventExecutorGroup;
    /**
     * 事件监听器
     */
    private final ChannelEventListener channelEventListener;

    private AtomicBoolean init = new AtomicBoolean(false);

    public NettyRemotingClient(NettyClientConfig nettyClientConfig) {
        this(nettyClientConfig, null);
    }

    public NettyRemotingClient(NettyClientConfig nettyClientConfig, ChannelEventListener channelEventListener) {
        this.bootstrap = new Bootstrap();
        this.eventLoopGroup = new NioEventLoopGroup(1, new ThreadFactoryImpl("NettyClientSelector_"));
        this.nettyClientConfig = nettyClientConfig;
        this.publicExecutor = buildPublicExecutor();
        this.channelEventListener = channelEventListener;

        this.defaultEventExecutorGroup = new DefaultEventExecutorGroup(nettyClientConfig.getClientWorkerThreads(),
            new ThreadFactoryImpl("NettyClientWorkerThread_"));

        bootstrap.group(eventLoopGroup).channel(useEpoll() ? EpollSocketChannel.class : NioSocketChannel.class)
            .handler(new LoggingHandler(LogLevel.INFO))
            .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, nettyClientConfig.getConnectTimeoutMillis())
            .option(ChannelOption.TCP_NODELAY, true) //
            .option(ChannelOption.SO_KEEPALIVE, true) //
            .handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    ch.pipeline().addLast(defaultEventExecutorGroup,
                        new IdleStateHandler(0, 0, nettyClientConfig.getClientChannelMaxIdleTimeSeconds()),
                        buildDecoder(), // 解码
                        buildEncoder(), // 编码
                        new NettyConnectManageHandler(), // 连接
                        new NettyClientHandler() // 数据处理
                    );
                }
            });
    }

    /**
     * 连接
     * 
     * @Author wangwei
     * @Date 2024/2/27
     */
    @SneakyThrows
    public Channel doConnect(InetSocketAddress inetSocketAddress) {
        init.compareAndSet(false, true);
        CompletableFuture<Channel> completableFuture = new CompletableFuture<>();
        bootstrap.connect(inetSocketAddress).addListener((ChannelFutureListener)future -> {
            if (future.isSuccess()) {
                log.info("The client has connected [{}] successful!", inetSocketAddress.toString());
            }
            completableFuture.complete(future.channel());
        });
        return completableFuture.get(nettyClientConfig.getConnectTimeoutMillis(), TimeUnit.MILLISECONDS);
    }

    public void shutdown() {
        if (publicExecutor != null) {
            publicExecutor.shutdown();
        }
        if (eventLoopGroup != null) {
            eventLoopGroup.shutdownGracefully();
        }
        clearRPCHook();
    }

    public void registerProcessor(int requestCode, NettyRequestProcessor processor, ExecutorService executor) {
        ExecutorService executorThis = executor;
        if (executor == null) {
            executorThis = publicExecutor;
        }
        Pair<NettyRequestProcessor, ExecutorService> pair = new Pair<>(processor, executorThis);
        this.processorTable.put(requestCode, pair);
    }

    @Override
    public ExecutorService getCallbackExecutor() {
        return this.publicExecutor;
    }

    @Override
    public ChannelEventListener getChannelEventListener() {
        return channelEventListener;
    }

    private ExecutorService buildPublicExecutor() {
        int publicThreadNums = nettyClientConfig.getClientCallbackExecutorThreads();
        if (publicThreadNums <= 0) {
            publicThreadNums = 4;
        }
        return Executors.newFixedThreadPool(publicThreadNums, new ThreadFactoryImpl("NettyClientPublicExecutor_"));
    }

    /**
     * 执行一次，无返回
     * @Author wangwei
     * @Date 2024/2/28
     */
    public void invokeOneway(Channel channel, RemotingCommand request) throws RemotingSendRequestException {
        if (channel != null && channel.isActive()) {
            this.invokeOnewayImpl(channel, request);
        } else {
            this.closeChannel(channel);
            String addr = RemotingHelper.parseChannelRemoteAddr(channel);
            throw new BusinessException("channel未注册或未激活，{}", addr);
        }
    }
    
    /**
     * 同步执行
     *
     * @param request 请求内容
     * @return RemotingCommand
     */
    public RemotingCommand invokeSync(Channel channel, RemotingCommand request)
        throws RemotingSendRequestException, RemotingTimeoutException, InterruptedException {
        if (channel != null && channel.isActive()) {
            return this.invokeSyncImpl(channel, request, this.nettyClientConfig.getRequestTimeoutMillis());
        } else {
            this.closeChannel(channel);
            String addr = RemotingHelper.parseChannelRemoteAddr(channel);
            throw new BusinessException("channel未注册或未激活，{}", addr);
        }
    }

    /**
     *
     * @param request 请求内容
     * @param requestTimeoutMillis 超时时间
     * @return
     */
    public RemotingCommand invokeSync(Channel channel, RemotingCommand request, Long requestTimeoutMillis)
        throws RemotingSendRequestException, RemotingTimeoutException, InterruptedException {
        if (channel != null && channel.isActive()) {
            return this.invokeSyncImpl(channel, request, requestTimeoutMillis);
        } else {
            this.closeChannel(channel);
            String addr = RemotingHelper.parseChannelRemoteAddr(channel);
            throw new BusinessException("channel未注册或未激活，{}", addr);
        }
    }

    /**
     * 异步执行
     *
     * @param request 请求内容
     * @param invokeCallback
     */
    public void invokeAsync(Channel channel, RemotingCommand request, InvokeCallback invokeCallback)
        throws RemotingConnectException {
        if (channel != null && channel.isActive()) {
            this.invokeAsyncImpl(channel, request, this.nettyClientConfig.getRequestTimeoutMillis(), invokeCallback);
        } else {
            this.closeChannel(channel);
            String addr = RemotingHelper.parseChannelRemoteAddr(channel);
            throw new BusinessException("channel未注册或未激活，{}", addr);
        }
    }

    /**
     *
     * @param request 请求内容
     * @param invokeCallback 回调内容
     * @param requestTimeoutMillis 超时时间
     * @throws RemotingConnectException
     */
    public void invokeAsync(Channel channel, RemotingCommand request, InvokeCallback invokeCallback, Long requestTimeoutMillis) {
        if (channel != null && channel.isActive()) {
            this.invokeAsyncImpl(channel, request, requestTimeoutMillis, invokeCallback);
        } else {
            this.closeChannel(channel);
            String addr = RemotingHelper.parseChannelRemoteAddr(channel);
            throw new BusinessException("channel未注册或未激活，{}", addr);
        }
    }

    /**
     * 异步执行
     * 
     * @param request 请求内容
     * @return CompletableFuture
     */
    public CompletableFuture<RemotingCommand> invokeAsyncFuture(Channel channel, RemotingCommand request) {
        CompletableFuture<RemotingCommand> future = new CompletableFuture<>();
        try {
            this.invokeAsync(channel, request, new InvokeCallback() {
                @Override
                public void operationComplete(ResponseFuture responseFuture) {

                }

                @Override
                public void operationSucceed(RemotingCommand response) {
                    future.complete(response);
                }

                @Override
                public void operationFail(Throwable throwable) {
                    future.completeExceptionally(throwable);
                }
            });
        } catch (RemotingConnectException e) {
            future.completeExceptionally(e);
        }
        return future;
    }

    private boolean useEpoll() {
        return NetworkUtil.isLinuxPlatform() && nettyClientConfig.isUseEpollNativeSelector() && Epoll.isAvailable();
    }

    class NettyClientHandler extends SimpleChannelInboundHandler<RemotingCommand> {

        @Override
        protected void channelRead0(ChannelHandlerContext ctx, RemotingCommand msg) throws Exception {
            processRequestCommand(ctx, msg);
        }
    }

    @ChannelHandler.Sharable
    public class NettyConnectManageHandler extends ChannelDuplexHandler {
        @Override
        public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
            final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
            log.info("NETTY CLIENT PIPELINE: channelRegistered {}", remoteAddress);
            super.channelRegistered(ctx);
        }

        @Override
        public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
            final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
            log.info("NETTY CLIENT PIPELINE: channelUnregistered, the channel[{}]", remoteAddress);
            super.channelUnregistered(ctx);
        }

        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
            log.info("NETTY CLIENT PIPELINE: channelActive, the channel[{}]", remoteAddress);
            super.channelActive(ctx);
            if (NettyRemotingClient.this.channelEventListener != null) {
                NettyRemotingClient.this
                    .putNettyEvent(new NettyEvent(NettyEventType.CONNECT, remoteAddress, ctx.channel()));
            }
        }

        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
            log.info("NETTY CLIENT PIPELINE: channelInactive, the channel[{}]", remoteAddress);
            super.channelInactive(ctx);
            if (NettyRemotingClient.this.channelEventListener != null) {
                NettyRemotingClient.this
                    .putNettyEvent(new NettyEvent(NettyEventType.CLOSE, remoteAddress, ctx.channel()));
            }
        }

        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
            if (evt instanceof IdleStateEvent) {
                IdleStateEvent event = (IdleStateEvent)evt;
                if (event.state().equals(IdleState.ALL_IDLE)) {
                    final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
                    log.warn("NETTY CLIENT PIPELINE: IDLE exception [{}]", remoteAddress);
                    RemotingHelper.closeChannel(ctx.channel());
                    if (NettyRemotingClient.this.channelEventListener != null) {
                        NettyRemotingClient.this
                            .putNettyEvent(new NettyEvent(NettyEventType.IDLE, remoteAddress, ctx.channel()));
                    }
                }
            }
            ctx.fireUserEventTriggered(evt);
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
            log.warn("NETTY CLIENT PIPELINE: exceptionCaught {}", remoteAddress);
            log.warn("NETTY CLIENT PIPELINE: exceptionCaught exception.{}", ExceptionUtils.getStackTrace(cause));
            if (NettyRemotingClient.this.channelEventListener != null) {
                NettyRemotingClient.this
                    .putNettyEvent(new NettyEvent(NettyEventType.EXCEPTION, remoteAddress, ctx.channel()));
            }

            RemotingHelper.closeChannel(ctx.channel());
        }
    }

    /**
     * 主动关闭Channel
     */
    public void closeChannel(Channel channel) {
        if (channel != null) {
            // 主动关闭
            String addr = RemotingHelper.parseChannelRemoteAddr(channel);
            log.info("通道未活跃，主动关闭：{}", addr);
            RemotingHelper.closeChannel(channel);
        } else {
            log.info("通道已关闭");
        }
    }
}
